package server

import (
	"fmt"
	"github.com/streadway/amqp"
	"log"
)

type Rabbitmq struct {
	option RabbitmqOption
	conn   *amqp.Connection
	ch     *amqp.Channel
	done   <-chan struct{}
}

type RabbitmqOption struct {
	Port     int
	Username string
	Password string
	Host     string
	Queue    string
}

func (d *Rabbitmq) Callback(task *Task) {
	log.Printf("rabbitmq callback running")
	ch, err := d.conn.Channel()
	if err != nil {
		log.Printf("channel publish error:%s", err)
		return
	}
	q, err := ch.QueueDeclare(d.option.Queue+"_CALLBACK", true, false, false, false, nil)
	if err != nil {
		log.Printf("queue declare error: %s", err)
	}
	err = ch.Publish(
		"",
		q.Name,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        task.Payload,
		},
	)
	if err != nil {
		log.Printf("channel publish error:%s", err)
		return
	}
	log.Printf("rabbitmq callbacked")
}

func (d *Rabbitmq) Start(platform *Platform) error {
	log.Printf("%s rabbitmq serve start", platform.Name)
	option := d.option
	conn, err := amqp.Dial(fmt.Sprintf("amqp://%s:%s@%s:%d/", option.Username, option.Password, option.Host, option.Port))
	if err != nil {
		return err
	}
	d.conn = conn
	ch, err := conn.Channel()
	if err != nil {
		return err
	}
	d.ch = ch
	q, err := ch.QueueDeclare(option.Queue, true, false, false, false, nil)
	if err != nil {
		return err
	}

	msg, err := ch.Consume(q.Name, "", true, false, false, false, nil)
	if err != nil {
		return err
	}
	go func() {
		for m := range msg {
			task := newTaskFromMessage(m)
			platform.AddTask(task)
		}
	}()
	return nil
}

func (d *Rabbitmq) Stop() {

}

func newTaskFromMessage(m amqp.Delivery) *Task {
	return &Task{
		Name:    "test",
		Delay:   1,
		Payload: []byte("this is rabbitmq payload"),
	}
}

func NewRabbitmq(option RabbitmqOption) (*Rabbitmq, error) {
	return &Rabbitmq{
		option: option,
		done:   make(chan struct{}),
	}, nil
}

func RabbitmqOptionSetter(option RabbitmqOption) OptionCall {
	return func(platform *Platform) {
		platform.SetOption(RABBITMQ, option)
	}
}
